package com.lagou.rpc.consumer.client;


import org.apache.catalina.Server;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.UnknownHostException;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;


@Component
public class RpcClientCluster implements ServerChangeProcessor {
    private List<RpcClient> rpcClientList = new ArrayList<>();

    private AtomicInteger reqCount = new AtomicInteger(0);

    private ReentrantLock lock = new ReentrantLock();

    private Random random;

    @Autowired
    private ClientConfig clientConfig;

    @Autowired
    private Registry registry;

    @Autowired
    private ConsumerReporter reporter;

    public void init() throws InterruptedException {
        this.random = new Random();

        try {
            registry.init();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }

        List<ServerInfo> serverInfos = registry.getServerInfos();
        updateRpcClients(serverInfos);

        try {
            registry.startWatch(this);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onServerChange(List<ServerInfo> serverInfos) {
        System.out.println("onServerChange:" + serverInfos);
        updateRpcClients(serverInfos);
    }

    //第一个版本(轮询)
    public RpcClient getRpcClient() {
        lock.lock();

        try {
            if (rpcClientList.size() == 0) {
                throw new RuntimeException("Server temporarily unavailable");
            }

            if (ClientConfig.ALG_ROUND_ROBIN.equals(clientConfig.getAlgorithm())) {
                //轮询
                int index = reqCount.incrementAndGet() % rpcClientList.size();
                return rpcClientList.get(index);
            }
            else {
                //按权重，在更新RPC客户端列表的时候已经排过序,随机获取一个
                int searchIndex = 0;
                while (searchIndex < (rpcClientList.size() - 1)) {
                    if (rpcClientList.get(searchIndex).getServerInfo().getLastTimeElapsed()
                        == rpcClientList.get(searchIndex + 1).getServerInfo().getLastTimeElapsed()) {
                        searchIndex++;
                    }
                    else { break; }
                }

                int index = random.nextInt(searchIndex + 1);
                return rpcClientList.get(index);
            }
        }
        finally {
            lock.unlock();
        }
    }

    private void updateRpcClients(List<ServerInfo> serverInfos) {
        lock.lock();

        try {
            Iterator<RpcClient> it = rpcClientList.iterator();
            while (it.hasNext()) {
                it.next();
                it.remove();
            }

            //如果是按权重做负载均衡,则根据上次调用时间对被调用的服务器信息进行排序
            if (ClientConfig.ALG_ROUND_ROBIN.equals(clientConfig.getAlgorithm())
                    && serverInfos != null && serverInfos.size() > 0
            ) {
                System.out.println("===update client list===before sort");
                for (ServerInfo serverInfo : serverInfos) {
                    System.out.println(serverInfo);
                }

                Collections.sort(serverInfos, new Comparator<ServerInfo>() {
                    @Override
                    public int compare(ServerInfo o1, ServerInfo o2) {
                        if (o1.getLastTimeElapsed() > o2.getLastTimeElapsed()) return 1;
                        else if (o1.getLastTimeElapsed() < o2.getLastTimeElapsed()) return -1;
                        else return 0;
                    }
                });

                System.out.println("===update client list===after sort");
                for (ServerInfo serverInfo : serverInfos) {
                    System.out.println(serverInfo);
                }
            }

            for (ServerInfo serverInfo : serverInfos) {
                RpcClient rpcClient = new RpcClient(serverInfo.getIp(),
                        serverInfo.getPort(),
                        serverInfo.getLastTimeElapsed(),
                        reporter
                );
                rpcClientList.add(rpcClient);
            }

            System.out.println("=====updateRpcClients=====");
            for (RpcClient rpcClient : rpcClientList) {
                System.out.println(rpcClient.getServerInfo());
            }
            System.out.println("=====end updateRpcClients=====");
        }
        finally {
            lock.unlock();
        }
    }

//    public static void main(String[] args) {
//        Random r = new Random();
//        System.out.println(r.nextInt(1));
//    }
}
